package com.song.elasticsearch.config.kafka;

import lombok.Data;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Data
@Configuration
@ConditionalOnBean(KafkaSecurity.class)
@ConfigurationProperties(prefix = "kafka.consumer")
public class KafkaConsumerConfig {
    private String bootstrapServers;
    private String groupId;
    private String keyDeserializer;
    private String valueDeserializer;
    private String securityProtocol;
    private String saslKerberosServiceName;
    private String kerberosDomainName;
    private String autoOffsetReset;
    private String autoCommitIntervalMs;
    private String enableAutoCommit;
    private String sessionTimeoutMs;

    private Boolean consumerSign = false;

    @Autowired
    KafkaSecurity kafkaSecurity;

    @Bean("kafkaConsumer")
    public KafkaConsumer<String, String> kafkaConsumer(){
        if (!consumerSign){
            kafkaSecurity.securityValidate();
            consumerSign = true;
        }
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(producerProperties());
        return kafkaConsumer;
    }

    public Properties producerProperties(){
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
        // 服务名
        properties.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, saslKerberosServiceName);
        // 域名
        properties.put(SaslConfigs.KERBEROS_DOMAIN_NAME, kerberosDomainName);
        //消费策略
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
        return properties;
    }
}
